<?php

namespace App\Core\Event;

use App\Core\Event\HllEvent;
use App\Events\Message\CommonMessage;
use App\Models\Table\HllEventTable;
use App\Repositories\HllEventRepository;
use App\Utils\AmqpConnection;
use Doctrine\ORM\EntityManager;
use Monolog\Logger;
use PhpAmqpLib\Exchange\AMQPExchangeType;
use PhpAmqpLib\Message\AMQPMessage;

class EventMessageProducer
{

    /**
     * @var HllEvent|null
     */
    private $event;

    public static $messageSequenceNumber = 1;

    public function __construct($event = null)
    {
        $this->event = $event;
    }

    /**
     * @param CommonMessage $abstractMessage ;
     * @throws IncorrectEventDefineException
     * @since v2.3 现在由事件的action 来作为publish的routekey
     * @since v2.4 支持通用的消息生产
     * @since v2.4.2 增加串码序列号 防止冲突
     */
    public function produce($abstractMessage = null)
    {
        if ($abstractMessage) {
            return $this->produceCommonMessage($abstractMessage);
        }
        if (!$this->event->getAction()) {
            throw new IncorrectEventDefineException('you should define the action name!');
        }
        try {
            $hllevent = new \App\Entities\HllEvent();
            $hllevent->setAction($this->event->getAction());
            /** @var AbstractMessage $message */
            $message = $this->event->getPlayload();
            $uniMessageId = $message->getId() . '-' .
                str_pad(self::$messageSequenceNumber, 2, 0, STR_PAD_LEFT);
            $message->setId($uniMessageId);
            $this->event->setPlayload($message); //更改了 message对象需要重新set
            $hllevent->setData(json_encode($this->event));
            $hllevent->setMessageId($message->getId());
            $hllevent->setBizId($message->getBizId());
            $hllevent->setCreatedAt(date('Y-m-d H:i:s'));
            $hllevent->setUpdatedAt(date('Y-m-d H:i:s'));
            $hllevent->setExchange($this->event->getExchange());
            $hllevent->setTransport($this->event->getTransport());
            $hllevent->setErrorTasks("");
            $hllevent->setErrorInfo("");
            $hllEventTable = new HllEventTable();
            $hllEventEntity = $hllEventTable->newEntityFromPopo($hllevent);
            $hllEventTable->save($hllEventEntity);
            if ($this->event->getTransport() == 'amqp') {
                //消息必须在落库之后发送，否则可能存在找不到的情况
                /** @var AmqpConnection $amqp */
                $amqp = app(AmqpConnection::class);
                $conn = $amqp->connect();
                $channel = $conn->channel();
                $body = json_encode($this->event);
                $msg = new AMQPMessage($body, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]);
                $msg->set("message_id",$message->getId());
                //消息 动态 绑定 路由键
                $channel->basic_publish($msg, $this->event->getExchange(), $this->event->getAction());
            }
            self::$messageSequenceNumber++;
        } catch (\Throwable $exception) {
            /** @var Logger $logger */
            $logger = app('logger');
            $logger->error('生产异步消息失败', ['msg' => $exception->getMessage()]);
        }
    }

    private function produceCommonMessage(CommonMessage $message)
    {

        if ($message->getTransport() == 'amqp') {
            /** @var AmqpConnection $amqp */
            $amqp = app(AmqpConnection::class);
            $conn = $amqp->connect();
            $channel = $conn->channel();
            $body = json_encode($message);
            $msg = new AMQPMessage($body, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]);
            //消息 动态 绑定 路由键
            $channel->basic_publish($msg, $message->getExchange(), $message->getAction());
        }

    }
}